package com.example.combine;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * Created with IntelliJ IDEA.
 * ClassName: UnionDemp
 * Package: com.example.combine
 * Description:
 * User: fzykd
 *
 * @Author: LQH
 * Date: 2023-07-19
 * Time: 9:55
 */

public class ConnectKeyByDemo {
    public static void main(String[] args) throws Exception {
        //需求:连接两条流，输出能根据id匹配上的数据（类似inner join效果）
        //笛卡尔积的效果

        //1.创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        //缓存不能做清理
        DataStreamSource<Tuple2<Integer, String>> source1 = env.fromElements(
                Tuple2.of(1, "a1"),
                Tuple2.of(1, "a2"),
                Tuple2.of(2, "b"),
                Tuple2.of(3, "c")
        );
        DataStreamSource<Tuple3<Integer, String, Integer>> source2 = env.fromElements(
                Tuple3.of(1, "aa1", 1),
                Tuple3.of(1, "aa2", 2),
                Tuple3.of(2, "bb", 1),
                Tuple3.of(3, "cc", 1)
        );

        ConnectedStreams<Tuple2<Integer, String>, Tuple3<Integer, String, Integer>> connect = source1.connect(source2);

        //多并行下 需要根据 关联条件进行keyby 才能保证相同的数据到一起 才能匹配的上
        ConnectedStreams<Tuple2<Integer, String>, Tuple3<Integer, String, Integer>>
                keyby = connect.keyBy(s1 -> s1.f0, s2 -> s2.f0);


        //和CoMapFunction类似
        SingleOutputStreamOperator<String> process = keyby.process(new CoProcessFunction<Tuple2<Integer, String>, Tuple3<Integer, String, Integer>, String>() {
            //1.要实现互相匹配的效果
            //两条流不一定谁的数据先来
            //每条流 有数据来 存在变量中(有状态)
            //每条数据来了之后 除了存在变量中 去另一条流的变量中 查找是否有匹配上的
            Map<Integer, List<Tuple2<Integer, String>>> s1Cache = new HashMap<>();
            Map<Integer, List<Tuple3<Integer, String, Integer>>> s2Cache = new HashMap<>();
            @Override
            public void processElement1(Tuple2<Integer, String> value, Context ctx, Collector<String> out) throws Exception {
                //s1的数据来就存在变量中 存在HashMap中 id=key  value=>List<>
                Integer id = value.f0;
                if (!s1Cache.containsKey(id)) {
                    //这是判断 如果key不存在 说明是第一条数据 初始化 put进map中
                    List<Tuple2<Integer, String>> arr1 = new ArrayList<>();
                    arr1.add(value);
                    s1Cache.put(id, arr1);
                } else {
                    //走到这里 说明不是第一条数据 key是存在的 直接添加到val的list中
                    s1Cache.get(id).add(value);
                }
                //存完之后 去另一条流查找 是否有匹配上的
                if (s2Cache.containsKey(id)) {
                    //如果有匹配的id
                    for (Tuple3<Integer, String, Integer> s2Element : s2Cache.get(id)) {
                        out.collect("s1: " + value + "------------" + "s2: " + s2Element);
                    }
                }
            }

            @Override
            public void processElement2(Tuple3<Integer, String, Integer> value, Context ctx, Collector<String> out) throws Exception {
                //s1的数据来就存在变量中 存在HashMap中 id=key  value=>List<>
                Integer id = value.f0;
                if (!s2Cache.containsKey(id)) {
                    //这是判断 如果key不存在 说明是第一条数据 初始化 put进map中
                    List<Tuple3<Integer, String, Integer>> arr2 = new ArrayList<>();
                    arr2.add(value);
                    s2Cache.put(id, arr2);
                } else {
                    //走到这里 说明不是第一条数据 key是存在的 直接添加到val的list中
                    s2Cache.get(id).add(value);
                }

                //存完之后 去另一条流查找 是否有匹配上的
                if (s1Cache.containsKey(id)) {
                    //如果有匹配的id
                    for (Tuple2<Integer, String> s1Element : s1Cache.get(id)) {
                        out.collect("s1: " + s1Element + "------------" + "s2: " + value);
                    }
                }

            }
        });

        process.print();

        env.execute();

    }
}
